/*
 * Copyright (c) 2018-2023 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "tree.h"
#include "js.h"

#ifdef __cplusplus
extern "C" {
#endif

static struct treeNode *get_user_node(HASH_TABLE_T *, struct job *);
static int update_borrowed_slots(struct treeNode *, struct job *, int, int);

static struct share_acct *make_sacct(const char *, unsigned int);
static LIST_T *make_usershare_list(const char *);
static void add_member_2list(const char *, unsigned int, struct uhgroup **, LIST_T *);
static unsigned int compute_slots(struct treeNode *, unsigned int, unsigned int);
static void sort_siblings(struct treeNode *, int (*cmp)(const void *, const void *));
static void sort_tree_byshares(struct tree *);
static void sort_tree_bydeserve(struct tree *);
static void reset_sacct_sent(struct tree *);

static int compare_sacct_node(const void *, const void *);

/**
 * @brief       policy: create fairshare tree according to queue definition
 *
 * @param[in|out]   fsqueue  #1: fairshare queue
 * @param[in]       num      #2: number of user group
 * @param[in]       ugroups  #3: user group data
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        This function should be called after reading qp.queues and qp.users.
 *              If failed to create fairshare tree for the queue, fairshare
 *              scheduling policy should be disabled.
 */
int
create_fairshare_tree(struct queue *fsqueue, unsigned int num, struct uhgroup **ugroups)
{
    if (!(fsqueue->schedule_policy & QUEUE_POLICY_FAIRSHARE)) {
        log_message(ERR, NOCLASS, "%s: queue <%s> is not a fairshare queue.",
                    __func__, fsqueue->queue_name);
        return -1;
    }

    LIST_T *acctNodeList;
    acctNodeList = gf_list_make("share account node list");

    struct tree *sacctTree;
    sacctTree = tree_create("");

    /* root summarizes all the tree counters */
    sacctTree->root->data = (struct share_acct *)make_sacct("/", 1);

    LIST_T *userShareList;
    userShareList = make_usershare_list(fsqueue->fairshare);
    if (NULL == userShareList) {
        log_message(ERR, NOCLASS, "%s: queue <%s> fairshare <%s> has syntax error.",
                    __func__, fsqueue->queue_name, fsqueue->fairshare);
        return -1;
    }

    struct treeNode *sacctNode=NULL, *root=NULL;
    while (1) {

        if (root) {
            add_member_2list(sacctNode->name, num, ugroups, userShareList);
        } else {
            root = sacctTree->root;
        }

        LIST_ITERATOR_T iter;
        gf_listiterator_link(&iter, userShareList);

        LIST_PROXY_T *shareProxy;
        for (shareProxy = (LIST_PROXY_T *)gf_listiterator_curentry(&iter);
             !gf_listiterator_isend(&iter);
             gf_listiterator_getnext(&iter, (LIST_T **)&shareProxy)) {

            struct share_acct *sacct;
            sacct = (struct share_acct *)shareProxy->subject;

            sacctNode = (struct treeNode *)calloc(1, sizeof(struct treeNode));
            if (NULL == sacctNode) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct treeNode));
                return -1;
            }

            sacctNode->name = sacct->name;
            sacctNode = tree_insert_node(root, sacctNode);

            sacctNode->data = sacct;

            LIST_PROXY_T *pxy;
            pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
            if (NULL == pxy) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct list_proxy));
                return -1;
            }

            pxy->subject = (void *)sacctNode;
            gf_list_insert(acctNodeList, acctNodeList, (LIST_T *)pxy);
        }

        /* Sort by shares so the tree is always sorted by share priority */
        sort_siblings(root, compare_sacct_node);

        gf_list_free(userShareList, NULL);
        userShareList = gf_list_make("user share list");

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)gf_list_pop(acctNodeList);
        if (NULL == pxy) {
            gf_list_free(userShareList, NULL);
            break;
        }

        sacctNode = (struct treeNode *)pxy->subject;
        root = sacctNode;
        FREEUP(pxy);
    }

    gf_list_free(acctNodeList, NULL);

    sacctNode = sacctTree->root;
    while ((sacctNode = tree_next_node(sacctNode))) {

        char buf[MAXNAMELEN+1];
        struct share_acct *sacct;
        sacct = (struct share_acct *)sacctNode->data;

        /* Create the hash table of nodes and their immediate parent. */
        if (NULL == sacctNode->child) {

	    /* all and default are synonyms as they
	     * both indicate a user that is not explicitly defined. */
            if (strcasecmp(sacct->name, ALL_STRING) == 0
		|| strcasecmp(sacct->name, DEFAULT_STRING) == 0) {
                sacct->options |= SACCT_USER_ALL;
	    }

            sacct->options |= SACCT_USER;
            snprintf(buf, MAXNAMELEN, "%s/%s", sacctNode->parent->name, sacctNode->name);
            gf_hash_install(sacctTree->node_table, buf, sacctNode);
        } else {
            sacct->options |= SACCT_GROUP;
        }
        snprintf(buf, MAXNAMELEN, "%s", sacctNode->name);
        gf_hash_install(sacctTree->node_table, buf, sacctNode);
    }

    sort_tree_byshares(sacctTree);

    fsqueue->fairshare_tree = sacctTree;

    return 0;
} // end function create_fairshare_tree

/**
 * @brief       policy: update user fairshare accouting information
 *
 * @param[in]   fsqueue      #1: fairshare queue
 * @param[in]   fsjob        #2: job
 * @param[in]   numpend      #3: number of pending slot of the job
 * @param[in]   numrun       #4: number of running slot of the job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        This function should be called after fairshare job status
 *              is changed.
 *              numpend is negative when job status is changed from pend to run.
 *              numrun is negative when job status is changed from run to done.
 */
int
update_fairshare_acct(struct queue *fsqueue, struct job *fsjob,
                      int numpend, int numrun)
{
    struct tree *queueTree=NULL;
    if (fsqueue->schedule_policy & QUEUE_POLICY_FAIRSHARE) {
        queueTree = fsqueue->fairshare_tree;
    } else if (fsqueue->schedule_policy & QUEUE_POLICY_OWNERSHIP) {
        queueTree = fsqueue->ownership_tree;
    }

    if (NULL == queueTree) {
        return -1;
    }

    struct treeNode *userNode;
    userNode = get_user_node(queueTree->node_table, fsjob);
    if (NULL == userNode) {
        return -1;
    }

    struct share_acct *sacct;
    sacct = (struct share_acct *)userNode->data;

    if (0 > numpend && 0 < numrun) {
        sacct->sent--;
    }

    if (update_borrowed_slots(userNode, fsjob, numpend, numrun)) {
        return 0;
    }

    int numRAN = 0;
    if (0 < numrun) {
        numRAN = numrun;
    }

    while (userNode) {
        sacct = (struct share_acct *)userNode->data;
        sacct->numPEND = sacct->numPEND + numpend;
        sacct->numRUN = sacct->numRUN + numrun;
        sacct->numRAN = sacct->numRAN + numRAN;
        userNode = userNode->parent;
    }

    return 0;
} // end function update_fairshare_acct

/**
 * @brief       policy: distribue fairshare slot to user fairshare account
 *
 * @param[in]   fsqueue      #1: fairshare queue
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, queue has not fairshare tree
 *
 * @note        This function should be called before scheduling session begins.
 */
int
fairshare_session_begin(struct queue *fsqueue)
{
    struct tree *queueTree=NULL;
    if (fsqueue->schedule_policy & QUEUE_POLICY_FAIRSHARE) {
        queueTree = fsqueue->fairshare_tree;
    } else if (fsqueue->schedule_policy & QUEUE_POLICY_OWNERSHIP) {
        queueTree = fsqueue->ownership_tree;
    }

    if (NULL == queueTree) {
        return -1;
    }

    LIST_T *acctNodeList;
    acctNodeList = gf_list_make("share account node list");

    LIST_PROXY_T *proxy;
    while (NULL != (proxy = (LIST_PROXY_T *)gf_list_pop(queueTree->leaf_list))) {
        FREEUP(proxy);
    }

    sort_tree_byshares(queueTree);
    reset_sacct_sent(queueTree);

    /* only after sort get the first child */
    struct treeNode *sacctNode;
    sacctNode = queueTree->root->child;

    int avail, nslot;
    avail = nslot = fsqueue->num_fairshare_slot;
    int tried = 0;
    while (1) {

        struct share_acct *sacct;

        /* Iterate at each tree level but don't traverse branches without tokens. */
        while (NULL != sacctNode) {

            sacct = (struct share_acct *)sacctNode->data;
            /* all is a dummy share account the tree is populated with real ones. */
            if (sacct->options & SACCT_USER_ALL) {
                sacctNode = sacctNode->right;
                continue;
            }

            /* enqueue as we want to traverse the tree by priority */
            if (NULL != sacctNode->child) {

                LIST_PROXY_T *pxy;
                pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
                if (NULL == pxy) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct list_proxy));
                    return -1;
                }

                pxy->subject = (void *)sacctNode;
                gf_list_insert(acctNodeList, acctNodeList, (LIST_T *)pxy);
            }

            avail = avail - compute_slots(sacctNode, nslot, avail);

            /* As we traverse in priority order the leafs are also sorted */
            if (NULL == sacctNode->child) {

                LIST_PROXY_T *pxy;
                pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
                if (NULL == pxy) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct list_proxy));
                    return -1;
                }

                pxy->subject = sacctNode;
                gf_list_insert(queueTree->leaf_list, queueTree->leaf_list, (LIST_T *)pxy);
            }

            sacctNode = sacctNode->right;
        }

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)gf_list_pop(acctNodeList);
        if (NULL != pxy) {

            sacctNode = (struct treeNode *)pxy->subject;
            FREEUP(pxy);

            /* tokens come from the parent */
            sacct = (struct share_acct *)sacctNode->data;
            avail = nslot = sacct->sent;
            sacctNode = sacctNode->child;

            continue;
        }

        if (0 < avail && 0 == tried) {

            ++ tried;
            sacctNode = queueTree->root->child;
            continue;
        }

        break;
    }

    gf_list_free(acctNodeList, NULL);

    return 0;
} // end function fairshare_session_begin

/**
 * @brief       policy: choose a job from a fairshare user who has both
 *                      slot share and pending job
 *
 * @param[in]   fsqueue      #1: fairshare queue
 * @param[out]  proxy        #2: proxy of choosed job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        This function should be called during scheduling session.
 *              Scheduler will try to dispatch the job if there is enough
 *              resource for the job to run.
 */
int
fairshare_choose_job(struct queue *fsqueue, struct jobProxy **proxy)
{
    struct tree *queueTree=NULL;
    if (fsqueue->schedule_policy & QUEUE_POLICY_FAIRSHARE) {
        queueTree = fsqueue->fairshare_tree;
    } else if (fsqueue->schedule_policy & QUEUE_POLICY_OWNERSHIP) {
        queueTree = fsqueue->ownership_tree;
    }

    if (NULL == queueTree) {
        return -1;
    }

    LIST_T *leaves;
    leaves = queueTree->leaf_list;
    if (0 == LIST_NUM_ENTS(leaves)) {
        *proxy = NULL;
        return -1;
    }

    struct share_acct *sacct;
    unsigned int numAcct;
    struct jobProxy *jref = NULL;
    struct job *fsJob;

    while (TRUE) {

        numAcct = 0;

        struct treeNode *userNode;
        LIST_PROXY_T *userProxy;
        while (NULL != (userProxy = (LIST_PROXY_T *)gf_list_pop(leaves))) {

            userNode = (struct treeNode *)userProxy->subject;
            FREEUP(userProxy);

            sacct = (struct share_acct *)userNode->data;
            if (0 < sacct->sent) {
                ++ numAcct;
                break;
            }
        }

        if (0 == numAcct) {
            *proxy = NULL;
            return -1;
        }

        log_message(DEBUG, FAIRSHARE, "%s: account %s num slots %d queue %s", __func__, sacct->name,
                    sacct->sent+1, fsqueue->queue_name);

        struct user *fairUser;
        fairUser = (struct user *)gf_hash_find(ghashUser, sacct->name);
        if (NULL == fairUser) {
            log_message(ERR, NOCLASS, "%s: user <%s> is not used by the cluster.",
                        __func__, sacct->name);
            continue;
        }

        bool_t found = FALSE;
        unsigned int count = 0;
        jref = NULL;
        fsJob = NULL;

        LIST_ITERATOR_T iter;
        gf_listiterator_link(&iter, fairUser->jobs);

        LIST_PROXY_T *jobProxy, *nextProxy;
        for (jobProxy = (LIST_PROXY_T *)gf_listiterator_curentry(&iter);
             !gf_listiterator_isend(&iter); jobProxy = nextProxy) {

            gf_listiterator_getnext(&iter, (LIST_T **)&nextProxy);

            ++count;

            jref = (struct jobProxy *)jobProxy->subject;
            fsJob = jref->schedule_job;

            if (fsJob->job_queue == fsqueue) {

                gf_list_remove(fairUser->jobs, (LIST_T *)jobProxy);
                found = TRUE;

                break;
            }
        }

        if (NULL == fsJob || !found) {

            log_message(DEBUG, FAIRSHARE, "%s: user account <%s> is chosen %d in queue %s but has no jobs.",
                        __func__, sacct->name, sacct->sent+1, fsqueue->queue_name);
            continue;
        }

        char strJobId[32];
        if ((NULL != fsJob) && (gf_get_logclass() & FAIRSHARE)) {

            log_message(DEBUG, FAIRSHARE, "%s: user account <%s> job %s queue %s found in %d iterations",
                        __func__, sacct->name, jobid_2string(fsJob->jobid, strJobId),
                        fsqueue->queue_name, count);
        }

        userProxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
        if (NULL == userProxy) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", sizeof(struct list_proxy));
            return -1;
        }

        userProxy->subject = userNode;
        gf_list_insert(leaves, leaves->forw, (LIST_T *)userProxy);

        *proxy = jref;

        break;
    }

    return 0;
} // end function fairshare_choose_job

int
fairshare_session_end(struct queue *fsqueue)
{
    return 0;
}

/**
 * @brief       policy: get the queue fairshare user account
 *
 * @param[in]   fsqueue      #1: fairshare queue
 * @param[out]  num          #2: number of user account
 * @param[out]  sacct        #3: fairshare user account
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        This function is called when client asks for queue's fairshare
 *              account information.
 */
int
fairshare_get_accts(struct queue *fsqueue, unsigned int *num, struct share_acct ***sacct)
{
    struct tree *queueTree=NULL;
    if (fsqueue->schedule_policy & QUEUE_POLICY_FAIRSHARE) {
        queueTree = fsqueue->fairshare_tree;
    } else if (fsqueue->schedule_policy & QUEUE_POLICY_OWNERSHIP) {
        queueTree = fsqueue->ownership_tree;
    }

    if (NULL == queueTree) {
        return -1;
    }

    // count the number of nodes
    struct treeNode *userNode;
    unsigned int numNode;
    userNode = queueTree->root;
    numNode = 0;
    while (NULL != (userNode = tree_next_node(userNode))) {
        ++numNode;
    }

    if (0 == numNode) {
        return 0;
    }

    *sacct = (struct share_acct **)calloc(numNode, sizeof(struct share_acct *));
    if (NULL == *sacct) {
        return -1;
    }

    // add the array elemens to the nodes
    numNode = 0;
    userNode = queueTree->root;
    while (NULL != (userNode = tree_next_node(userNode))) {

        struct share_acct *userAcct;
        userAcct = (struct share_acct *)(userNode->data);
        (*sacct)[numNode] = (struct share_acct *)calloc(1, sizeof(struct share_acct));
        if (NULL == (*sacct)[numNode]) {
            return -1;
        }

        memcpy((*sacct)[numNode], userAcct, sizeof(struct share_acct));
        (*sacct)[numNode]->name = copy_string(userAcct->name);
        numNode ++;
    }

    *num = numNode;

    return 0;
} // end function fairshare_get_accts

/**
 * @brief       policy: create ownership tree according to queue definition
 *
 * @param[in|out]   ownqueue #1: ownership queue
 * @param[in]       num      #2: number of user group
 * @param[in]       ugroups  #3: user group data
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        This function should be called after reading qp.queues and qp.users.
 *              If failed to create ownership tree for the queue, ownership
 *              scheduling policy should be disabled.
 */
int
create_ownership_tree(struct queue *ownqueue, unsigned int num, struct uhgroup **ugroups)
{
    if (!(ownqueue->schedule_policy & QUEUE_POLICY_OWNERSHIP)) {

        log_message(ERR, NOCLASS, "%s: ownership scheduling for queue <%s> is not successfully initialized.",
                    __func__, ownqueue->queue_name);
        return -1;
    }

    LIST_T *acctNodeList;
    acctNodeList = gf_list_make("share account node list");

    struct tree *sacctTree;
    sacctTree = tree_create("");

    /* root summarizes all the tree counters */
    sacctTree->root->data = (struct share_acct *)make_sacct("/", 1);

    LIST_T *userShareList;
    userShareList = make_usershare_list(ownqueue->ownership);
    if (NULL == userShareList) {
        log_message(ERR, NOCLASS, "%s: queue <%s> ownership <%s> has syntax error.",
                    __func__, ownqueue->queue_name, ownqueue->ownership);
        return -1;
    }

    struct treeNode *sacctNode=NULL, *root=NULL;
    while (1) {

        if (root) {
            add_member_2list(sacctNode->name, num, ugroups, userShareList);
        } else {
            root = sacctTree->root;
        }

        LIST_ITERATOR_T iter;
        gf_listiterator_link(&iter, userShareList);

        LIST_PROXY_T *shareProxy;
        for (shareProxy = (LIST_PROXY_T *)gf_listiterator_curentry(&iter);
             !gf_listiterator_isend(&iter);
             gf_listiterator_getnext(&iter, (LIST_T **)&shareProxy)) {

            struct share_acct *sacct;
            sacct = (struct share_acct *)shareProxy->subject;

            sacctNode = (struct treeNode *)calloc(1, sizeof(struct treeNode));
            if (NULL == sacctNode) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct treeNode));
                return -1;
            }

            sacctNode->name = sacct->name;
            sacctNode = tree_insert_node(root, sacctNode);

            sacctNode->data = sacct;

            LIST_PROXY_T *pxy;
            pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
            if (NULL == pxy) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct list_proxy));
                return -1;
            }

            pxy->subject = (void *)sacctNode;
            gf_list_insert(acctNodeList, acctNodeList, (LIST_T *)pxy);
        }

        /* Sort by shares so the tree is always sorted by share priority */
        sort_siblings(root, compare_sacct_node);

        gf_list_free(userShareList, NULL);
        userShareList = gf_list_make("user share list");

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)gf_list_pop(acctNodeList);
        if (NULL == pxy) {
            gf_list_free(userShareList, NULL);
            break;
        }

        sacctNode = (struct treeNode *)pxy->subject;
        root = sacctNode;
        FREEUP(pxy);
    }

    gf_list_free(acctNodeList, NULL);

    sacctNode = sacctTree->root;
    while ((sacctNode = tree_next_node(sacctNode))) {

        char buf[MAXNAMELEN+1];
        struct share_acct *sacct;
        sacct = (struct share_acct *)sacctNode->data;

        /* Create the hash table of nodes and their immediate parent. */
        if (NULL == sacctNode->child) {

	    /* all and default are synonyms as they
	     * both indicate a user that is not explicitly defined. */
            if (strcasecmp(sacct->name, ALL_STRING) == 0
		|| strcasecmp(sacct->name, DEFAULT_STRING) == 0) {
                sacct->options |= SACCT_USER_ALL;
	    }

            sacct->options |= SACCT_USER;
            snprintf(buf, MAXNAMELEN, "%s/%s", sacctNode->parent->name, sacctNode->name);
            gf_hash_install(sacctTree->node_table, buf, sacctNode);
        } else {
            sacct->options |= SACCT_GROUP;
        }
        snprintf(buf, MAXNAMELEN, "%s", sacctNode->name);
        gf_hash_install(sacctTree->node_table, buf, sacctNode);
    }

    sort_tree_byshares(sacctTree);

    ownqueue->ownership_tree = sacctTree;

    return 0;
} // end function create_ownership_tree

/**
 * @brief       policy: distribue ownership slot to user fairshare account
 *
 * @param[in]   ownqueue     #1: ownership queue
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, queue has not fairshare tree
 *
 * @note        This function should be called before scheduling session begins.
 */
int
ownership_session_begin(struct queue *ownqueue)
{
    if (0 == ownqueue->num_ownership_slot) {
        return 0;
    }

    struct tree *ownershipTree;
    ownershipTree = ownqueue->ownership_tree;

    /* Distribute the tokens all the way down the leaves */
    LIST_T *acctNodeList;
    acctNodeList = gf_list_make("share account node list");

    LIST_PROXY_T *proxy;
    while (NULL != (proxy = (LIST_PROXY_T *)gf_list_pop(ownershipTree->leaf_list))) {
        FREEUP(proxy);
    }

    sort_tree_bydeserve(ownershipTree);
    reset_sacct_sent(ownershipTree);

    /* only after sort get the first child */
    struct treeNode *sacctNode;
    sacctNode = ownershipTree->root->child;

    int avail, nslot;
    avail = nslot = ownqueue->num_ownership_slot;

    int first=0, tried=0;

    while (1) {

        struct share_acct *sacct;

        /* Iterate at each tree level but
         * don't traverse branches without tokens. */
        while (NULL != sacctNode) {

            sacct = (struct share_acct *)sacctNode->data;
            /* all is a dummy share account the tree is populated with real ones. */
            if (sacct->options & SACCT_USER_ALL) {
                sacctNode = sacctNode->right;
                continue;
            }

            /* enqueue as we want to traverse the tree by priority */
            if (NULL != sacctNode->child) {

                LIST_PROXY_T *pxy;
                pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
                if (NULL == pxy) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct list_proxy));
                    return -1;
                }

                pxy->subject = (void *)sacctNode;
                gf_list_insert(acctNodeList, acctNodeList, (LIST_T *)pxy);
            }

            /* Enforce the ownership/guarantee at the first level of the tree. */
            if (0 == first) {
                if (sacct->numRUN > sacct->shares) {
                    sacct->sent = 0;
                } else {
                   sacct->sent = sacct->shares - sacct->numRUN;
                }
            } else {
                avail = avail - compute_slots(sacctNode, nslot, avail);
            }

            /* As we traverse in priority order the leafs are also sorted */
            if (NULL == sacctNode->child) {

                LIST_PROXY_T *pxy;
                pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
                if (NULL == pxy) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct list_proxy));
                    return -1;
                }

                pxy->subject = sacctNode;
                gf_list_insert(ownershipTree->leaf_list, ownershipTree->leaf_list, (LIST_T *)pxy);
            }

            sacctNode = sacctNode->right;
        }

        ++first;

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)gf_list_pop(acctNodeList);
        if (NULL != pxy) {

            sacctNode = (struct treeNode *)pxy->subject;
            FREEUP(pxy);

            /* tokens come from the parent */
            sacct = (struct share_acct *)sacctNode->data;
            avail = nslot = sacct->sent;
            sacctNode = sacctNode->child;

            continue;
        }

        if (0 < avail && 0 == tried) {

            ++ tried;
            sacctNode = ownershipTree->root->child;
            continue;
        }

        break;
    }

    gf_list_free(acctNodeList, NULL);

    return 0;
} // end function ownership_session_begin

/**
 * @brief       get user node from fairshare user account tree
 *
 * @param[in|out]   node_tab #1: user account hash table
 * @param[in]       fsjob    #2: job
 *
 * @retval      not NULL     #1: succeed
 * @retval      NULL         #2: failed
 *
 * @note        Get the fairshare user account node for updating its account
 *              information. If not in hash table, try to add one.
 */
static struct treeNode *
get_user_node(HASH_TABLE_T *node_tab, struct job *fsjob)
{
    char key[MAXNAMELEN+1];

    if (NULL != fsjob->shared->submit.user_group
        && '\0' != fsjob->shared->submit.user_group[0]) {

        snprintf(key, MAXNAMELEN, "%s/%s", fsjob->shared->submit.user_group, fsjob->user_name);
    } else {
        snprintf(key, MAXNAMELEN, "%s", fsjob->user_name);
    }

    struct treeNode *userNode;
    userNode = (struct treeNode *)gf_hash_find(node_tab, key);
    if (NULL != userNode) { // found the user account node
        return userNode;
    }

    if (!is_clusteradmin(fsjob->user_name)
        && !isuser_queuemember(fsjob->user_name, fsjob->job_queue)) {
        // user is not the member of queue, do not update fairshare tree. 
        return NULL;
    }

    // If job specifies user group lookup, try to find it.
    if (NULL != fsjob->shared->submit.user_group
        && '\0' != fsjob->shared->submit.user_group[0]) {

        snprintf(key, MAXNAMELEN, "%s/all", fsjob->shared->submit.user_group);
        userNode = (struct treeNode *)gf_hash_find(node_tab, key);
    } else {
        // If all/default in fairshare account tree?
        userNode = (struct treeNode *)gf_hash_find(node_tab, ALL_STRING);
        if (NULL == userNode) {
            userNode = (struct treeNode *)gf_hash_find(node_tab, DEFAULT_STRING);
        }
    }

    if (NULL == userNode) { // should not update fairshare account tree
        return NULL;
    }

    struct share_acct *sacct;
    sacct = (struct share_acct *)userNode->data;
  
    struct treeNode *newUserNode;
    newUserNode = (struct treeNode *)calloc(1, sizeof(struct treeNode));
    if (NULL == newUserNode) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct treeNode));
        return NULL;
    }

    struct share_acct *sacct2;
    sacct2 = make_sacct(fsjob->user_name, sacct->shares);
    newUserNode->data = sacct2;
    newUserNode->name = copy_string(fsjob->user_name);

    tree_insert_node(userNode->parent, newUserNode);

    sacct2->options |= SACCT_USER;
    snprintf(key, MAXNAMELEN, "%s/%s", newUserNode->parent->name, newUserNode->name);
    gf_hash_install(node_tab, key, newUserNode);
    snprintf(key, MAXNAMELEN, "%s", newUserNode->name);
    gf_hash_install(node_tab, key, newUserNode);

    unsigned int sum=0;

    newUserNode = userNode->parent->child;
    while (NULL != newUserNode) {

        sacct = (struct share_acct *)newUserNode->data;
        if (sacct->options & SACCT_USER_ALL) {
            newUserNode = newUserNode->right;
            continue;
        }
        sum = sum + sacct->shares;
        newUserNode = newUserNode->right;
    }

    newUserNode = userNode->parent->child;
    while (NULL != newUserNode) {

        sacct = (struct share_acct *)newUserNode->data;
        if (sacct->options & SACCT_USER_ALL) {
            newUserNode = newUserNode->right;
            continue;
        }

        sacct->dshares = (double)sacct->shares/(double)sum;
        newUserNode = newUserNode->right;
    }

    return userNode->parent->child;
} // end function get_user_node

/**
 * @brief       update borrowed slot of the fairshare user account.
 *
 * @param[in]   usernode     #1: user account
 * @param[in]   osjob        #2: job
 * @param[in]   numpend      #3: number of pending slot of the job
 * @param[in]   numrun       #4: number of running slot of the job
 *
 * @retval      TRUE         #1: succeed
 * @retval      FALSE        #2: failed
 */
static int
update_borrowed_slots(struct treeNode *usernode, struct job *osjob,
                      int numpend, int numrun)
{
    // No ownership policy
    if (!(osjob->job_queue->schedule_policy & QUEUE_POLICY_OWNERSHIP)) {
        return FALSE;
    }

    if (0 == numrun) {
        return FALSE;
    } else if (0 > numrun && !(osjob->job_tag & JFLAG_BORROWED_SLOTS)) {
        return FALSE;
    }

    struct share_acct *sacct;
    if (0 < numrun) {
        struct treeNode *loopNode;

        loopNode = usernode;
        while (loopNode->parent->parent) {
            loopNode = loopNode->parent;
        }

        sacct = (struct share_acct *)loopNode->data;
        if (sacct->numRUN < sacct->shares) {
            return FALSE;
        }
    }

    // initialize numRAN in case of negative numrun
    int numRAN = 0;
    if (0 < numrun) {
        numRAN = numrun;
    }

    sacct = (struct share_acct *)usernode->data;
    while (usernode) {
        sacct = (struct share_acct *)usernode->data;
        sacct->numPEND = sacct->numPEND + numpend;
        sacct->numBORROWED = sacct->numBORROWED + numrun;
        sacct->numRAN = sacct->numRAN + numRAN;
        usernode = usernode->parent;
    }

    // mark/unmark job borrowed slot flag
    if (0 < numrun) {
        osjob->job_tag |= JFLAG_BORROWED_SLOTS;
    } else if (0 >numrun && (osjob->job_tag & JFLAG_BORROWED_SLOTS)) {
        osjob->job_tag &= ~JFLAG_BORROWED_SLOTS;
    }

    return TRUE;
} // end function update_borrowed_slots

static void
tokenize(char *p)
{
    unsigned int len;
    len = strlen(p);

    unsigned int i;
    for (i=0; i<len; i++) {
        if ('[' == p[i] || ']' == p[i] || ',' == p[i]) {
            p[i] = ' ';
        }
    }

    return;
}

static LIST_T *
make_usershare_list(const char *usershare)
{
    char *strUserShares;
    strUserShares = copy_string(usershare);

    char *p;
    p = strchr(strUserShares, '[');
    *p++ = 0;

    tokenize(p);

    LIST_T *shareList;
    shareList = gf_list_make("user share list");

    unsigned int sum=0;
    while (1) {

        char name[128];
        unsigned int shares;
        int cc, n;
        cc = sscanf(p, "%s%u%n", name, &shares, &n);
        if (cc == EOF) {
            break;
        }

        if (cc != 2) {
            free(strUserShares);
            return NULL;
        }

        p = p + n;

        struct share_acct *sacct;
        sacct = make_sacct(name, shares);

        sum += sacct->shares;

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
        if (NULL == pxy) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", sizeof(struct list_proxy));
            return NULL;
        }

        pxy->subject = (void *)sacct;
        gf_list_insert(shareList, shareList, (LIST_T *)pxy);
    }

    LIST_ITERATOR_T iter;
    gf_listiterator_link(&iter, shareList);

    LIST_PROXY_T *pxy;
    for (pxy = (LIST_PROXY_T *)gf_listiterator_curentry(&iter);
         !gf_listiterator_isend(&iter);
         gf_listiterator_getnext(&iter, (LIST_T **)&pxy)) {

        struct share_acct *sacct;
        sacct = (struct share_acct *)pxy->subject;
        sacct->dshares = (double)sacct->shares/(double)sum;
    }

    FREEUP(strUserShares);

    return shareList;
} // end function make_usershare_list

struct share_acct *
make_sacct(const char *name, unsigned int shares)
{
    struct share_acct *sacct;

    sacct = (struct share_acct *)calloc(1, sizeof(struct share_acct));
    if (NULL == sacct) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct share_acct));
        return NULL;
    }

    sacct->name = copy_string(name);
    sacct->shares = shares;

    return sacct;
}

static struct share_acct *
get_sacct(const char *acctname, const char *user_list)
{
    char *strUserList, *strHead;
    strHead = strUserList = copy_string(user_list);

    char name[128];
    unsigned int shares;

    int cc, n;
    cc = sscanf(strUserList, "%s%u%n", name, &shares, &n);
    if (cc == EOF) {
        free(strHead);
        return NULL;
    }

    struct share_acct *sacct=NULL;
    /* default can have users all or explicitly named users */
    if (strcmp(name, "default") == 0) {

        sacct = make_sacct(acctname, shares);
        free(strHead);
        return sacct;
    }

    /* Match the user in the user_list */
    while (1) {

        cc = sscanf(strUserList, "%s%u%n", name, &shares, &n);
        if (cc == EOF)
            break;
        if (strcmp(name, acctname) != 0) {
            strUserList = strUserList + n;
            continue;
        }

        sacct = make_sacct(name, shares);
        break;
    }

    free(strHead);
    return sacct;
}

static void
add_member_2list(const char *gname, unsigned int num, struct uhgroup **ugroups,
                 LIST_T *sharelist)
{
    struct uhgroup *gdata=NULL;
    char *strSaveUserShare=NULL;

    unsigned int i;
    for (i=0; i<num; i++) {

	/* Match the group name and the group must have shares. */
        if (NULL == ugroups[i]->user_shares) {
            continue;
        }

        if (0 != strcmp(gname, ugroups[i]->group_name)) {
            continue;
        }

        gdata = ugroups[i];

        strSaveUserShare = ugroups[i]->user_shares;
        gdata->user_shares = copy_string(strSaveUserShare);
        tokenize(gdata->user_shares);

        break;
    }

    if (NULL == gdata) {
        return;
    }

    unsigned int sum = 0;

    char *strMember;
    HASH_WALK_T walk;
    char *key;
    gf_hash_walk_start(gdata->hash_member, &walk);
    while (NULL != (strMember = (char *)gf_hash_walk(&walk, &key))) {

        struct share_acct *sacct;
        sacct = get_sacct(strMember, gdata->user_shares);
        if (NULL == sacct) {
            continue;
        }

        sum = sum + sacct->shares;

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
        if (NULL == pxy) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", sizeof(struct list_proxy));
            return;
        }

        pxy->subject = (void *)sacct;
        gf_list_insert(sharelist, sharelist, (LIST_T *)pxy);
    }

    LIST_ITERATOR_T iter;
    gf_listiterator_link(&iter, sharelist);

    LIST_PROXY_T *pxy;
    for (pxy = (LIST_PROXY_T *)gf_listiterator_curentry(&iter);
         !gf_listiterator_isend(&iter);
         gf_listiterator_getnext(&iter, (LIST_T **)&pxy)) {

        struct share_acct *sacct;
        sacct = (struct share_acct *)pxy->subject;
        sacct->dshares = (double)sacct->shares/(double)sum;
    }

    FREEUP(gdata->user_shares);
    gdata->user_shares = strSaveUserShare;

    return;
} // end function add_member_2list

static void
sort_siblings(struct treeNode *root, int (*cmp)(const void *, const void *))
{
    if (NULL == root->child) {
        return;
    }

    int num = 0;

    struct treeNode *sacctNode;
    sacctNode = root->child;
    while (NULL != sacctNode) {
        ++num;
        sacctNode = sacctNode->right;
    }

    if (1 == num) {
        return;
    }

    struct treeNode **sacctNodeVector;
    sacctNodeVector = (struct treeNode **)calloc(num, sizeof(struct treeNode *));
    if (NULL == sacctNodeVector) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", num*sizeof(struct treeNode *));
        return;
    }

    sacctNode = root->child;
    num = 0;
    while (NULL != sacctNode) {
        tree_remove_node(sacctNode);
        sacctNodeVector[num] = sacctNode;
        ++num;
        sacctNode = sacctNode->right;
    }
    root->child = NULL;

    /* We want to sort in ascending order as we use
     * tree_insert_node() which always inserts
     * node in the left most position. */
    qsort(sacctNodeVector, num, sizeof(struct treeNode *), cmp);

    int i;
    for (i = 0; i < num; i++) {
        sacctNodeVector[i]->parent = NULL;
        sacctNodeVector[i]->right = NULL;
        sacctNodeVector[i]->left = NULL;
        tree_insert_node(root, sacctNodeVector[i]);
    }

    FREEUP(sacctNodeVector);

    return;
}

static int
compare_sacct_node(const void *x, const void *y)
{
    struct treeNode *sacctNode1;
    struct treeNode *sacctNode2;
    struct share_acct *sacct1;
    struct share_acct *sacct2;

    sacctNode1 = *(struct treeNode **)x;
    sacctNode2 = *(struct treeNode **)y;

    sacct1 = (struct share_acct *)sacctNode1->data;
    sacct2 = (struct share_acct *)sacctNode2->data;

    /* We want to sort in ascending order as we use tree_inser_node()
     * which always inserts node in the left most position. */
    if (sacct1->shares > sacct2->shares) {
        return 1;
    }

    if (sacct1->shares < sacct2->shares) {
        return -1;
    }

    return 0;
}

static void
make_leaves(struct tree *saccttree)
{
    LIST_PROXY_T *proxy;
    while (NULL != (proxy = (LIST_PROXY_T *)gf_list_pop(saccttree->leaf_list))) {
        FREEUP(proxy);
    }

    struct treeNode *sacctNode;
    sacctNode = saccttree->root;
    while ((sacctNode = tree_next_node(sacctNode))) {

        if (NULL == sacctNode->child) {

            LIST_PROXY_T *pxy;
            pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
            if (NULL == pxy) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct list_proxy));
                return;
            }

            pxy->subject = sacctNode;
            gf_list_insert(saccttree->leaf_list, saccttree->leaf_list, (LIST_T *)pxy);
        }
    }

    return;
}

static void
sort_tree_byshares(struct tree *saccttree)
{
    LIST_T *acctNodeList;
    acctNodeList = gf_list_make("share account node list");

    struct treeNode *root;
    root = saccttree->root;

    struct treeNode *sacctNode;
    sacctNode = root->child;

    while (1) {

        while (NULL != sacctNode) {

            if (NULL != sacctNode->child) {

                LIST_PROXY_T *pxy;
                pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
                if (NULL == pxy) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct list_proxy));
                    return;
                }

                pxy->subject = (void *)sacctNode;
                gf_list_insert(acctNodeList, acctNodeList, (LIST_T *)pxy);
            }

            sacctNode = sacctNode->right;
        }

        sort_siblings(root, compare_sacct_node);

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)gf_list_pop(acctNodeList);
        if (NULL == pxy) {
            break;
        }

        sacctNode = (struct treeNode *)pxy->subject;
        FREEUP(pxy);
        root = sacctNode;
        sacctNode = root->child;
    }

    gf_list_free(acctNodeList, NULL);
    make_leaves(saccttree);

    return;
}

static unsigned int
compute_slots(struct treeNode *sacctnode, unsigned int total, unsigned int avail)
{
    struct share_acct *sacct;
    sacct = (struct share_acct *)sacctnode->data;

    unsigned int slot;
    slot = (unsigned int)ceil(sacct->dshares * (double)total);

    slot = MIN(slot, avail);

    sacct->sent = sacct->sent + MIN(sacct->numPEND, slot);

    return sacct->sent;
}

static unsigned int
compute_deserve(struct treeNode *sacctnode, unsigned int sum, unsigned int avail)
{
    struct share_acct *sacct;
    sacct = (struct share_acct *)sacctnode->data;

    unsigned int value;
    value = (unsigned int)ceil(sacct->dshares * (double)sum);

    /* How much you deserve from what you should have used. */
    sacct->deserve = value - sacct->numRAN;

    return sacct->deserve;
}

static int
compare_sacct(const void *x, const void *y)
{
    struct treeNode *sacctNode1;
    struct treeNode *sacctNode2;
    struct share_acct *sacct1;
    struct share_acct *sacct2;

    sacctNode1 = *(struct treeNode **)x;
    sacctNode2 = *(struct treeNode **)y;

    sacct1 = (struct share_acct *)sacctNode1->data;
    sacct2 = (struct share_acct *)sacctNode2->data;

    /* Sort in descending order as the tree insertion algorithm always inserts right.
     * 5 2 -1 -4 will become -4 -1 2 5 once on the tree */
    int ret = 0;
    if (0 < sacct1->deserve && 0 > sacct2->deserve) {
        ret = 1;
    } else if (0 > sacct1->deserve && 0 < sacct2->deserve) {
        ret = -1;
    } else {
        if (sacct1->deserve > sacct2->deserve) {
            ret = 1;
        } else if (sacct1->deserve < sacct2->deserve) {
            ret = -1;
        }
    }

    return ret;
}

static void
sort_tree_bydeserve(struct tree *saccttree)
{
    LIST_T *acctNodeList;
    acctNodeList = gf_list_make("share account node list");

    struct treeNode *root;
    root = saccttree->root;

    struct treeNode *sacctNode;
    sacctNode = root->child;

    while (1) {

        struct treeNode *saveSacctNode;
        saveSacctNode = sacctNode;

        int sum = 0;
        while (NULL != sacctNode) {

            if (NULL != sacctNode->child) {

                LIST_PROXY_T *pxy;
                pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
                if (NULL == pxy) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct list_proxy));
                    return;
                }

                pxy->subject = (void *)sacctNode;
                gf_list_insert(acctNodeList, acctNodeList, (LIST_T *)pxy);
            }

            struct share_acct *sacct;
            sacct = (struct share_acct *)sacctNode->data;
            sacct->deserve = 0;

            /* sum up the historical. */
            sum = sum + sacct->numRAN;

            sacctNode = sacctNode->right;
        }

        int avail = sum;
        sacctNode = saveSacctNode;
        while (NULL != sacctNode) {

            avail = avail - compute_deserve(sacctNode, sum, avail);
            sacctNode = sacctNode->right;
        }

        sort_siblings(root, compare_sacct);

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)gf_list_pop(acctNodeList);
        if (NULL == pxy) {
            break;
        }

        sacctNode = (struct treeNode *)pxy->subject;
        FREEUP(pxy);

        root = sacctNode;
        sacctNode = sacctNode->child;
    }

    gf_list_free(acctNodeList, NULL);
    make_leaves(saccttree);

    return;
}

static void
reset_sacct_sent(struct tree *saccttree)
{
    struct treeNode *root;
    root = saccttree->root;

    struct treeNode *sacctNode;
    sacctNode = root->child;

    LIST_T *childList;
    childList = gf_list_make("share account list");

    while (1) {

        while (NULL != sacctNode) {

            if (NULL != sacctNode->child) {

                LIST_PROXY_T *pxy;
                pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
                if (NULL == pxy) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct list_proxy));
                    return;
                }

                pxy->subject = (void *)sacctNode;
                gf_list_insert(childList, childList, (LIST_T *)pxy);
            }

            struct share_acct *sacct;
            sacct = (struct share_acct *)sacctNode->data;
            sacct->sent = 0;
            sacctNode = sacctNode->right;
        }

        LIST_PROXY_T *pxy;
        pxy = (LIST_PROXY_T *)gf_list_pop(childList);
        if (NULL == pxy) {
            break;
        }

        sacctNode = (struct treeNode *)pxy->subject;
        FREEUP(pxy);

        root = sacctNode;
        sacctNode = root->child;
    }

    gf_list_free(childList, NULL);

    return;
}

#ifdef __cplusplus
}
#endif
